package com.smsc.headend.task.engine.service.impl;

import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.module.data.dto.TermProgressDTO;
import com.smsc.headend.module.meterscheme.entity.CollTermTaskPlan;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.task.engine.mapper.CollTermTaskPlanMapper;
import com.smsc.headend.task.engine.service.CollTermTaskPlanService;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

/**
 * @author xiaosc
 * @date 2021/04/09
 */
@Service
@Slf4j
public class CollTermTaskPlanServiceImpl extends ServiceImpl<CollTermTaskPlanMapper, CollTermTaskPlan> implements CollTermTaskPlanService {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    DataSource dataSource;

    @Value("${spring.datasource.dbtype:oracle}")
    String dbType;

    @Value("${spring.datasource.dbnames}")
    String[] dnNames;


    @Override
    public void saveTermCollectProgress(Long tv, Long termId, Long lpDataItemId) {
        TermProgressDTO progressDTO = new TermProgressDTO();
        progressDTO.setLpDataItemId(lpDataItemId);
        progressDTO.setTermId(termId);
        progressDTO.setProgressTv(tv);

        kafkaTemplate.send(TaskKafkaTopic.COLLECT_DCU_PROGRESS_PLAN_UPDATE, JSONUtil.toJsonStr(progressDTO).getBytes(Charset.forName("utf-8")));
    }

    @Override
    public Boolean batchSaveTermCollectProgress(List<TermProgressDTO> dtoList) {
        if ("oracle".equals(dbType)) {
            return batchSaveTermCollectProgressForOracle(dtoList);
        } else if ("postgresql".equals(dbType)) {
            return batchSaveTermCollectProgressForPostsql(dtoList);
        } else {
            log.error("unknown dbType");
            return false;
        }

    }

    private Boolean batchSaveTermCollectProgressForOracle(List<TermProgressDTO> dtoList) {
        String sql = "MERGE INTO COLL_TERM_TASK_PLAN  A USING (SELECT ? AS TERM_ID, ?  AS PROGRESS_TIME, ? AS MEASUREMENT_PROFILE_ID, ? AS UPDATE_TIME  FROM DUAL) B " +
                "ON (A.TERM_ID = B.TERM_ID AND A.MEASUREMENT_PROFILE_ID = B.MEASUREMENT_PROFILE_ID) " +
                "WHEN MATCHED THEN UPDATE SET A.PROGRESS_TIME = B.PROGRESS_TIME, A.UPDATE_TIME = B.UPDATE_TIME " +
                "WHEN NOT MATCHED THEN INSERT (TERM_ID, MEASUREMENT_PROFILE_ID, PROGRESS_TIME, UPDATE_TIME) VALUES ( B.TERM_ID, B.MEASUREMENT_PROFILE_ID, B.PROGRESS_TIME, B.UPDATE_TIME )";

        Long updateTv = SystemClock.now() / 1000;
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (TermProgressDTO progressDTO : dtoList) {
                ps.setLong(1, progressDTO.getTermId());
                ps.setLong(2, progressDTO.getProgressTv());
                ps.setLong(3, progressDTO.getLpDataItemId());
                ps.setLong(4, updateTv);
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }
        return true;
    }

    private Boolean batchSaveTermCollectProgressForPostsql(List<TermProgressDTO> dtoList) {
        String sql = "with upsert as (" +
                "    update COLL_TERM_TASK_PLAN A\n" +
                "        set PROGRESS_TIME = B.PROGRESS_TIME,UPDATE_TIME = B.UPDATE_TIME\n" +
                "        from (SELECT ? AS TERM_ID, ?  AS MEASUREMENT_PROFILE_ID, ? AS PROGRESS_TIME, ? AS UPDATE_TIME ) B\n" +
                "        where A.TERM_ID = B.TERM_ID AND A.MEASUREMENT_PROFILE_ID = B.MEASUREMENT_PROFILE_ID\n" +
                "        returning A.TERM_ID,A.MEASUREMENT_PROFILE_ID\n" +
                ")\n" +
                "insert\n" +
                "into COLL_TERM_TASK_PLAN\n" +
                "select *\n" +
                "from (SELECT ? AS TERM_ID, ?  AS MEASUREMENT_PROFILE_ID, ? AS PROGRESS_TIME, ? AS UPDATE_TIME) t\n" +
                "where not exists(\n" +
                "        select 1\n" +
                "        from upsert B\n" +
                "        where t.TERM_ID = B.TERM_ID\n" +
                "          and t.MEASUREMENT_PROFILE_ID = B.MEASUREMENT_PROFILE_ID\n" +
                "    );";

        Long updateTv = SystemClock.now() / 1000;
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);
        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (TermProgressDTO progressDTO : dtoList) {
                ps.setLong(1, progressDTO.getTermId());
                ps.setLong(2, progressDTO.getLpDataItemId());
                ps.setLong(3, progressDTO.getProgressTv());
                ps.setLong(4, updateTv);
                ps.setLong(5, progressDTO.getTermId());
                ps.setLong(6, progressDTO.getLpDataItemId());
                ps.setLong(7, progressDTO.getProgressTv());
                ps.setLong(8, updateTv);
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();

            ps.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }
        return true;
    }


}
